﻿using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQDemo.DelayQueue.Shared;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Volo.Abp.DependencyInjection;

namespace RabbitMQDemo.DelayQueue.Consumer01
{
   public  class MessageConsumer01Service : ITransientDependency
    {
        public void  Run()
        {
            using (var connection = new ConnectionFactory().CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    //string queueName = "zzhellodelay10s";
                    //channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);

                    string exchangeName = "exchange-direct-delay10s";
                    channel.ExchangeDeclare(exchange: exchangeName, type: "direct");

                    string name = channel.QueueDeclare().QueueName;
                    channel.QueueBind(queue: name, exchange: exchangeName, routingKey: "routing-delay-delay10s");

                    //channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: "routing-delay-delay10s");

                    //回调，当consumer收到消息后会执行该函数
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body.ToArray());
                        MessageEntity messageEntity = JsonConvert.DeserializeObject<MessageEntity>(message);
                        Console.WriteLine(" Consumer01 Received message:{0},time:{1}", messageEntity.Body, DateTime.Now);
                        bool succ = DoWork(messageEntity).Result;
                        if (!succ)
                        {
                            //执行失败，重新放入队列
                            Console.WriteLine("执行失败，重新入队，这次大概延迟执行20s");
                            ReQueue(messageEntity);
                        }
                    };

                    channel.BasicConsume(queue: name,
                               autoAck: true,
                               consumer: consumer);

                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
        }

        public async Task<bool> DoWork(MessageEntity messageEntity)
        {
            // 做一系列的业务逻辑，比如调用第三方接口，crud数据库等等。await doSth();
            if (new Random().Next(0, 10) == 1)
                return true;
            return false;
        }

        public void ReQueue(MessageEntity messageEntity)
        {
            var factory = new ConnectionFactory() { HostName = "127.0.0.1", UserName = "guest", Password = "guest" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    Dictionary<string, object> dic = new Dictionary<string, object>();
                    dic.Add("x-expires", 30000);
                    dic.Add("x-message-ttl", 20000);//队列上消息过期时间，应小于队列过期时间 
                    dic.Add("x-dead-letter-exchange", "exchange-direct-delay20s");//过期消息转向路由 
                    dic.Add("x-dead-letter-routing-key", "routing-delay-delay20s");//过期消息转向路由相匹配routingkey 
                                                                                   //创建一个名叫"zzhello"的消息队列
                    channel.QueueDeclare(queue: "zzhellodelay20s",
                      durable: true,
                      exclusive: false,
                      autoDelete: false,
                      arguments: dic);
                    messageEntity.RetryTimes += 1;
                    var message = JsonConvert.SerializeObject(messageEntity);
                    var body = Encoding.UTF8.GetBytes(message);
                    //向该消息队列发送消息message
                    channel.BasicPublish(exchange: "",
                      routingKey: "zzhellodelay20s",
                      basicProperties: null,
                      body: body);
                    Console.WriteLine(" Producer RsSent message:{0},time:{1}", message, DateTime.Now);
                }
            }
        }
    }
}
